package calao.storage.bdb;

import static calao.util.Logger.DEBUG;
import static calao.util.Logger.log;
import calao.storage.Index;
import calao.storage.Table;
import calao.storage.Field;

import com.sleepycat.bind.tuple.TupleInput;
import com.sleepycat.je.DatabaseEntry;

class RecordReader
{
	
	final Table table;
	protected Field field;
	protected int numFieldValues;
	protected int fieldValueNum;
	protected Object fieldValue;
	
	RecordReader(Table table)
	{
		this.table = table;
	}
	
	boolean readRecord(DatabaseEntry pkey, DatabaseEntry data)
	{
		if (readPrimaryKey(pkey)) return true;
		return readData(data);
	}
	
	boolean readPrimaryKey(DatabaseEntry pkey)
	{
		TupleInput keyIn = Util.toInput(pkey);
		Index index = table.getPrimaryKeyIndex();
		for (int i=0; i<index.getFieldCount(); i++) {
			field = index.getField(i);
			if (!skipsField()) {
				Object value = readValue(keyIn, -1);
				if (onRead(value)) return true;
			}
		}		
		return false;
	}
	
	boolean readData(DatabaseEntry data)
	{
		TupleInput in = Util.toInput(data);
		in.readPackedInt(); // encoding version
		while (in.available()>0) {
			int fieldId = in.readPackedInt();
			field = BdbStorage.envRegistry.getField(fieldId);
			numFieldValues = in.readPackedInt();
			if (field==null) {
				log(DEBUG, "Skipping unknown field %d", fieldId);
				for (int i=0; i<numFieldValues; i++) skipValueBlock(in);
			}
			else if (skipsField()) {
				for (int i=0; i<numFieldValues; i++) skipValueBlock(in);
			}
			else {
				fieldValueNum = 0;
				if (field.isMultiValued()) {
					Object[] values = field.getType().newValueArray(numFieldValues);
					for (; fieldValueNum<numFieldValues; fieldValueNum++) {
						if (readValueBlock(in)) return true;
						values[fieldValueNum] = fieldValue;
					}
					if (onRead(values)) return true;
				}
				else {
					if (numFieldValues!=0) {
						if (readValueBlock(in)) return true;
					}
					else fieldValue = null;
					if (onRead(fieldValue)) return true;
				}						
			}
		}
		return false;
	}
	
	/** 
	 * @return True to skip field.
	 */
	boolean skipsField()
	{
		return false;
	}
	
	/** 
	 * @return True to abort reading.
	 */
	boolean onRead(Object[] values) 
	{
		return false;
	}
	
	/** 
	 * @return True to abort reading.
	 */
	boolean onRead(Object value) 
	{ 
		return false;
	}
		
	/** Reads value block to {@link #fieldValue}. 
	 * Context-valued attributes:
	 * <ul>
	 * <li>{@link #field}: current field. 
	 * <li>{@link #numFieldValues}: total number of values for the current field. 
	 * <li>{@link #fieldValueNum}: position of the value to read.
	 * </ul> 
	 * @return True to abort reading.
	 */
	protected boolean readValueBlock(TupleInput in)
	{
		int size = readValueBlockLength(in);
		int startPos = in.getBufferOffset();
		fieldValue = readValue(in, size);
		int readCount = in.getBufferOffset()-startPos;
		if (readCount<size) in.skipFast(size-readCount);
		return false;
	}

	private static void skipValueBlock(TupleInput in) 
	{
		in.skipFast(readValueBlockLength(in));
	}
		
	private Object readValue(TupleInput in, int size)
	{
		switch (field.getType().toEnum()) {
		case INTEGER:
			return in.readInt();
		case LONG:
			return in.readLong();
		case SHORT:
			return in.readShort();
		case DOUBLE:
			return in.readDouble();
		case STRING:
			return in.readString();
		case OBJECT:
			try {
				TupleDataReader reader = new TupleDataReader(in, size);
				return reader.readObject();
			}
			catch (Exception e) {
				log(e);
			}
		}
		return null;
	}

	static int readValueBlockLength(TupleInput in)
	{
		int size = in.readUnsignedByte();
		if ((size&0x80)!=0) size &= 0x7F;
		else {
			size = Util.readInt(in.getBufferBytes(), in.getBufferOffset()-1);
			in.skipFast(3);
		}
		return size;
	}

}
